import pandas as pd
from clickhouse_driver import Client
from datetime import datetime, timedelta
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ClickHouseDailyNew:
    def __init__(self, host='localhost', port=9000, user='default', 
                 password='', database='db_cfx'):
        self.client = Client(
            host=host,
            port=port,
            user=user,
            password=password,
            database=database,
            settings={'use_numpy': True}  # 可选，提高性能
        )
        self.database = database
    
    def get_daily_new_data(self, start_date, end_date=None):
        """
        获取指定日期范围的日活数据
        
        Args:
            start_date: 开始日期 (YYYY-MM-DD 或 datetime)
            end_date: 结束日期，默认为昨天
            
        Returns:
            DataFrame: 包含日活数据的DataFrame
        """
        if end_date is None:
            end_date = datetime.now() - timedelta(days=1)
        
        # 转换日期格式
        if isinstance(start_date, str):
            start_date = datetime.strptime(start_date, '%Y-%m-%d')
        if isinstance(end_date, str):
            end_date = datetime.strptime(end_date, '%Y-%m-%d')
        
        query = """
        SELECT 
            base_create_date,
            count(DISTINCT uuid) as daily_active_users,
            count(*) as total_events
        FROM t_log_base_data_standard_active
        WHERE base_create_date BETWEEN %(start_date)s AND %(end_date)s
        GROUP BY base_create_date
        ORDER BY base_create_date
        """
        
        params = {
            'start_date': start_date.strftime('%Y-%m-%d'),
            'end_date': end_date.strftime('%Y-%m-%d')
        }
        
        try:
            result = self.client.query_dataframe(query, params)
            logger.info(f"成功获取 {len(result)} 天的日活数据")
            return result
        except Exception as e:
            logger.error(f"查询日活数据失败: {e}")
            return pd.DataFrame()
    
    def get_daily_new_detail(self, target_date):
        """
        获取指定日期的详细日活数据
        """
        if isinstance(target_date, str):
            target_date = datetime.strptime(target_date, '%Y-%m-%d')
        
        query = """
        select activate_date,
        count(),
        uniqExact(oaid)
        from db_cfx.t_data_user_cpa_active 
        WHERE activate_date = %(target_date)s 
        group by activate_date
        
        """
        
        params = {'target_date': target_date.strftime('%Y-%m-%d')}
        
        try:
            result = self.client.query_dataframe(query, params)
            logger.info(f"成功获取 {target_date} 的详细日活数据，共 {len(result)} 条记录")
            return result
        except Exception as e:
            logger.error(f"查询详细日活数据失败: {e}")
            return pd.DataFrame()
        
    def check_ch_new_data_completeness(self, target_date):
        """
        检查数据完整性（与目标表对比）
        """
        query = """
        WITH source_data AS (
            SELECT uniqExact(uuid)
            FROM db_cfx.t_log_base_data_standard_active as a
            left join db_cfx.t_data_user_cpa_new as b
            on a.uuid = b.oaid 
            WHERE toDate(base_create_time) = %(target_date)s and ((base_app_id='AnySelectedID' and vn>'3.37') or base_app_id!='AnySelectedID') and (b.oaid is null or b.oaid='')
        ),
        source_anyselectedid_data AS (
            SELECT uniqExact(uuid)
            FROM db_cfx.t_log_base_data_standard_active as a
            left join db_cfx.t_data_user_cpa_new as b
            on a.uuid = b.oaid 
            WHERE toDate(base_create_time) = %(target_date)s and (base_app_id='AnySelectedID' and vn>'3.37') and (b.oaid is null or b.oaid='')
        ),
        source_otrher_data AS (
            SELECT uniqExact(uuid)
            FROM db_cfx.t_log_base_data_standard_active as a
            left join db_cfx.t_data_user_cpa_new as b
            on a.uuid = b.oaid 
            WHERE toDate(base_create_time) = %(target_date)s and (base_app_id!='AnySelectedID') and (b.oaid is null or b.oaid='')
        ),
        target_data AS (
            SELECT DISTINCT oaid
            FROM t_data_user_cpa_active
            WHERE toDate(activate_time) = %(target_date)s
        ),
        target_asid_data AS (
            SELECT DISTINCT oaid
            FROM t_data_user_cpa_active
            WHERE toDate(activate_time) = %(target_date)s and channel='AnySelectedID'
        ),
        target_other_data AS (
            SELECT DISTINCT oaid
            FROM t_data_user_cpa_active
            WHERE toDate(activate_time) = %(target_date)s and channel!='AnySelectedID'
        )
        SELECT 
            (SELECT * FROM source_data) as expected_count,
            (SELECT * FROM source_anyselectedid_data) as expected_asid_count,
            (SELECT * FROM source_otrher_data) as expected_other_count,
            (SELECT count() FROM target_asid_data) as actual_asid_count,
            (SELECT count() FROM target_other_data) as actual_other_count,
            (SELECT count() FROM target_data) as actual_count,
            (SELECT count() FROM source_data) - (SELECT count() FROM target_data) as missing_count,
            (SELECT count() FROM target_data) * 100.0 / (SELECT count() FROM source_data) as completeness_ratio
        """
        
        params = {'target_date': target_date.strftime('%Y-%m-%d')}
        
        try:
            result = self.client.query_dataframe(query, params)
            return result.iloc[0] if len(result) > 0 else None
        except Exception as e:
            #logger.error(f"检查数据完整性失败: {e}")
            logger.error(f"检查数据完整性失败")
            return None
    
    def check_data_completeness(self, target_date):
        """
        检查数据完整性（与目标表对比）
        """
        query = """
        WITH mysql_data AS (
            select count() from db_cfx.t_mysql_data_user_cpa_active
            where toDate(activate_time/1000)=%(target_date)s
            group by toDate(activate_time/1000)
        ),
        ch_data AS (
            select count() from db_cfx.t_data_user_cpa_new final where activate_date=%(target_date)s group by activate_date
        ),
        ch_deduction_data AS (
            select count() from db_cfx.t_data_user_cpa_new_deduction final where activate_date=%(target_date)s group by activate_date
        ),
        ch_final_data AS (
            select count() from db_cfx.t_data_user_cpa_new_final final where activate_date=%(target_date)s group by activate_date
        )
        SELECT 
            (SELECT * FROM mysql_data) as mysql_new_count,
            (SELECT * FROM ch_data) as ch_new_count,
            (SELECT * FROM ch_final_data) as ch_final_count,
            (SELECT * FROM ch_deduction_data) as ch_deduction_count,
            (SELECT * FROM ch_data) - (SELECT * FROM ch_final_data) as deduct_miss_count,
            (SELECT * FROM mysql_data) - (SELECT * FROM ch_data) as missing_count,
            (SELECT * FROM ch_data) * 100.0 / (SELECT * FROM mysql_data) as completeness_ratio,
            ((SELECT * FROM ch_data)-(SELECT * FROM ch_final_data)) * 100.0 / (SELECT * FROM ch_data) as deduction_ratio
        """
        
        params = {'target_date': target_date.strftime('%Y-%m-%d')}
        
        try:
            result = self.client.query_dataframe(query, params)
            return result.iloc[0] if len(result) > 0 else None
        except Exception as e:
            #logger.error(f"检查数据完整性失败: {e.message}")
            logger.error(f"检查数据完整性失败")
            return None

# 使用示例
if __name__ == "__main__":
    # 创建连接实例
    ch = ClickHouseDailyNew(
        host='47.245.149.255',  # 您的ClickHouse服务器IP
        port=9000,
        user='default',
        password='Keepgo123@cys',  # 如果有密码
        database='db_cfx'
    )
    
    # 获取最近7天的日活数据
    end_date = datetime.now() - timedelta(days=1)
    start_date = end_date
    
    # daily_data = ch.get_daily_active_data(start_date, end_date)
    # print("日活数据概览:")
    # print(daily_data)
    
    # # # 获取某一天的详细数据
    
    # detail_data = ch.get_daily_active_detail(target_date)
    # print(f"\n{target_date.strftime('%Y-%m-%d')} 详细数据:")
    # print(detail_data.head())

    query_count=3
    while True:
        target_date = datetime.now() - timedelta(days=query_count)

        # 检查数据完整性
        # completeness = ch.check_ch_new_data_completeness(target_date)
        # if completeness is not None:
        #     print(f"\n{target_date.strftime('%Y-%m-%d')} ch原始日志和新增表数据完整性检查:")
        #     print(f"预期数量: {completeness['expected_count']}")
        #     print(f"实际数量: {completeness['actual_count']}")
        #     print(f"缺失数量: {completeness['missing_count']}")
        #     print(f"完整率: {completeness['completeness_ratio']:.2f}%")
        #     print(f"预期asid数量: {completeness['expected_asid_count']}")
        #     print(f"预期other数量: {completeness['expected_other_count']}")
        #     print(f"实际asid数量: {completeness['actual_asid_count']}")
        #     print(f"实际other数量: {completeness['actual_other_count']}")
        # 检查数据完整性
        completeness = ch.check_data_completeness(target_date)
        if completeness is not None:
            print(f"\n{target_date.strftime('%Y-%m-%d')} ch和mysql数据完整性检查:")
            print(f"Mysql新增数量: {completeness['mysql_new_count']}")
            print(f"Ch新增数量: {completeness['ch_new_count']}")
            print(f"缺失数量: {completeness['missing_count']}")
            print(f"完整率: {completeness['completeness_ratio']:.2f}%")

            print(f"Ch final新增数量: {completeness['ch_final_count']}")
            print(f"扣量表数量: {completeness['ch_deduction_count']}")
            print(f"扣量数量（new-final）: {completeness['deduct_miss_count']}")
            print(f"扣量比例: {(1-completeness['deduction_ratio']):.2f}%")
        query_count-=1
        if query_count<0:
            break